package com.tc.vms.config;

import com.tc.vms.iface.VmsServiceAdaptor;
import com.tc.vms.consumer.DlqQueueMetaData;
import com.tc.vms.consumer.QueueMetaData;
import com.tc.vms.exception.VmsClientErrorCode;
import com.tc.vms.exception.VmsClientException;
import com.tc.vms.iface.IContext;
import com.tc.vms.iface.VmsConnectionWrap;
import com.tc.vms.iface.VmsMetaData;
import com.tc.vms.log.VmsClientLogger;
import com.tc.vms.producer.ChannelMetaData;
import com.tc.vms.utils.Utils;
import com.tc.vms.utils.ZkBootOptions;
import com.tc.vms.utils.ZkUtils;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.slf4j.Logger;

import java.io.IOException;
import java.util.*;

/**
 * Created by yonghua.zhang on 2015/12/9.
 */
public class VmsRouterManager extends VmsServiceAdaptor {
    private final static Logger LOGGER = VmsClientLogger.getLog();
    private ZkClient vmsZkClient;
    private final VmsClientContext vmsClientContext;

    public VmsRouterManager(VmsClientContext vmsClientContext) {
        this.vmsClientContext = vmsClientContext;
    }

    public VmsClientContext getVmsClientContext() {
        return vmsClientContext;
    }

    public int getWatchersNum() {
        return vmsZkClient.numberOfListeners();
    }

    private synchronized ZkClient newVmsZkClientByBootstrapZk() throws VmsClientException {
        ZkBootOptions zkBootOptions = vmsClientContext.getZkBootOptions();
        ZkClient bootstrapZkClient = null;
        try {
            LOGGER.info("connect to the address of cfgcenter zookeeper [{}]", vmsClientContext.getBootstrapCfgCenter());
            bootstrapZkClient = ZkUtils
                    .createZkClient(vmsClientContext.getBootstrapCfgCenter(), zkBootOptions.getZKSessionTimeout(),
                            zkBootOptions.getZKConnectionTimeout());
            String bootstrapPath = vmsClientContext.getBootstrapNodePath();
            String vmsZkCenter = null;

            if (!ZkUtils.pathExists(bootstrapZkClient, vmsClientContext.getBootstrapChroot())) {
                LOGGER.error("The partition [{}] of cfgcenter is not exists", vmsClientContext.getBootstrapChroot());
                throw new VmsClientException(String.format("The partition [%s] of cfgcenter is not exists",
                        vmsClientContext.getBootstrapChroot()));
            }

            try {
                if (ZkUtils.pathExists(bootstrapZkClient, bootstrapPath)) {
                    Properties props = ZkUtils.readProperties(bootstrapZkClient, bootstrapPath);
                    vmsZkCenter = props.getProperty(vmsClientContext.getBootstrapNodeTag());
                }
            } catch (ZkNoNodeException ex) {
                LOGGER.info("The directory of [{}] not exists in Bootstrap zookeeper, vmszk will use [{}].",
                        bootstrapPath, vmsClientContext.getBootstrapCfgCenter());
            }

            if (vmsZkCenter == null || vmsZkCenter.isEmpty()) {
                LOGGER.info("connect to the address of vms zookeeper [{}]", vmsClientContext.getBootstrapCfgCenter());
                vmsZkClient = bootstrapZkClient;
            } else {
                LOGGER.info("connect to the address of vms zookeeper [{}]", vmsZkCenter);
                try {
                    vmsZkClient = ZkUtils.createZkClient(vmsZkCenter, zkBootOptions.getZKSessionTimeout(),
                            zkBootOptions.getZKConnectionTimeout());
                } catch (Exception ex) {
                    throw new VmsClientException("Route to vmsZkCenter failure", ex);
                } finally {
                    if (bootstrapZkClient != null) {
                        bootstrapZkClient.close();
                    }
                }
            }
        } catch (Exception e) {
            LOGGER.error("VmsRouterManager route to vms configcenter error.", e);
            throw new VmsClientException("VmsRouterManager route to vms configcenter error.",
                    VmsClientErrorCode.ZK_ERR_CODE_ROUTE_TO_EXCEPTION, e);
        }
        return vmsZkClient;
    }

    private Properties readZkNode2Props(String path) throws IOException {
        return ZkUtils.readProperties(vmsZkClient, path);
    }

    public Properties readZkChannelNodeProp(String channelName) throws IOException {
        return readZkNode2Props(vmsClientContext.getVmsZkChannelNodePath(channelName));
    }

    public Properties readZkQueueNodeProp(String queueName) throws IOException {
        return readZkNode2Props(vmsClientContext.getVmsZkQueueNodePath(queueName));
    }

    public Properties readZkClusterNodeProp(String clusterName) throws IOException {
        return readZkNode2Props(clusterName);
    }

    private String genVmsMonitorPath(VmsMetaData vmsMetaData) {
        String vmsMonitorPath = null;
        if (vmsMetaData instanceof ChannelMetaData) {
            //channel节点 zk watcher路径
            ChannelMetaData channelMetaData = (ChannelMetaData) vmsMetaData;
            vmsMonitorPath = vmsClientContext.getVmsZkChannelNodePath(channelMetaData.getChannelName());
        } else {
            //queue节点  zk watcher路径
            QueueMetaData queueMetaData = (QueueMetaData) vmsMetaData;
            vmsMonitorPath = vmsClientContext.getVmsZkQueueNodePath(queueMetaData.getQueueName());
        }
        return vmsMonitorPath;
    }

    public synchronized void registerWatcherHandler(VmsConnectionWrap vmsConnection) throws VmsClientException {
        try {
            VmsMetaData vmsMetaData = vmsConnection.getVmsMetaData();
            String vmsMonitorPath = genVmsMonitorPath(vmsMetaData);

            if (vmsMonitorPath != null && !vmsMonitorPath.isEmpty()) {
                LOGGER.debug("registerWatcherHandler -> Watching vms path = " + vmsMonitorPath);
                ZkUtils.ZKPath.registerDataChangeListener(vmsZkClient, vmsMonitorPath, vmsConnection);
            } else {
                LOGGER.warn("Register channel or queue failed, because the specified path [{}] is null.", vmsMetaData);
            }

            //monitor cluster path
            String clusterMonitorPath = ZkUtils.getVmsClusterPath(vmsMetaData.getCfgCenterProps());
            if (clusterMonitorPath != null && !clusterMonitorPath.isEmpty()) {
                LOGGER.debug("registerWatcherHandler -> Watching cluster path = " + clusterMonitorPath);
                ZkUtils.ZKPath.registerDataChangeListener(vmsZkClient, clusterMonitorPath, vmsConnection);
            } else {
                LOGGER.warn("Register cluster path failed, because the specified path [{}] is null.", vmsMetaData);
            }

            //register session monitor
            ZkUtils.ZKPath.registerStateChangeListener(vmsZkClient, vmsConnection);
        } catch (Exception e) {
            throw new VmsClientException("registerWatcherHandler exception.", e);
        }
    }

    public synchronized void unregisterWatcherHandler(VmsConnectionWrap vmsConnection) throws VmsClientException {
        try {
            VmsMetaData vmsMetaData = vmsConnection.getVmsMetaData();
            String vmsMonitorPath = genVmsMonitorPath(vmsMetaData);

            if (vmsMonitorPath != null && !vmsMonitorPath.isEmpty()) {
                LOGGER.debug("unregisterWatcherHandler -> Watching vms path = " + vmsMonitorPath);
                ZkUtils.ZKPath.unregisterDataChangeListener(vmsZkClient, vmsMonitorPath, vmsConnection);
            }

            //cluster path
            String clusterMonitorPath = ZkUtils.getVmsClusterPath(vmsMetaData.getCfgCenterProps());
            if (clusterMonitorPath != null && !clusterMonitorPath.isEmpty()) {
                LOGGER.debug("unregisterWatcherHandler -> Watching cluster path = " + clusterMonitorPath);
                ZkUtils.ZKPath.unregisterDataChangeListener(vmsZkClient, clusterMonitorPath, vmsConnection);
            }

            //unregister session monitor
            ZkUtils.ZKPath.unregisterStateChangeListener(vmsZkClient, vmsConnection);
        } catch (Exception e) {
            throw new VmsClientException("unregisterWatcherHandler exception.", e);
        }
    }

    public synchronized void registerInstEphemeralPath(VmsConnectionWrap vmsConnection) throws VmsClientException {
        VmsMetaData vmsMetaData = vmsConnection.getVmsMetaData();
        String instName = vmsMetaData.getClientInstName();
        if (instName != null && !instName.isEmpty()) {
            String tempZkPath = "";
            if (vmsMetaData instanceof ChannelMetaData) {
                tempZkPath = vmsClientContext.getVmsZkChannelNodePath(vmsMetaData.getZnodeName());

            } else if (vmsMetaData instanceof QueueMetaData) {
                tempZkPath = vmsClientContext.getVmsZkQueueNodePath(vmsMetaData.getZnodeName());
            }
            String instPath = Utils.mkZkPathString(tempZkPath, vmsMetaData.getClientInstName());

            tempZkPath = ZkUtils.ZKPath.createEphemeralSequential(vmsZkClient, instPath, "");
            vmsConnection.setZkEphemeralPath(tempZkPath);
            LOGGER.info("registerInstEphemeralPath the path of VmsMetaData is [{}]", tempZkPath);
        }
    }

    public synchronized void unregisterInstEphemeralPath(VmsConnectionWrap vmsConnection) throws VmsClientException {
        String zkEphemeralPath = vmsConnection.getZkEphemeralPath();
        if (zkEphemeralPath != null && !zkEphemeralPath.isEmpty()) {
            ZkUtils.ZKPath.delete(vmsZkClient, zkEphemeralPath);
        }
    }

    enum ZkNodeType {
        CHANNEL,
        QUEUE
    }

    private Properties getMergePropsFromZkNodePath(String name, ZkNodeType type) throws IOException {
        Properties props = null;
        try {
            switch (type) {
                case CHANNEL:
                    props = readZkChannelNodeProp(name);
                    break;
                case QUEUE:
                    props = readZkQueueNodeProp(name);
                    break;
            }
            Properties clusterProps = readZkClusterNodeProp(ZkUtils.getVmsClusterPath(props));
            Properties mergeProps = new Properties();
            mergeProps.putAll(clusterProps);
            mergeProps.putAll(props);
            return mergeProps;
        } catch (Exception ex) {
            throw new IOException("The znode of [" + name + "] is wrong or not exists.", ex);
        }

    }

    private Properties getMergePropsFromVmsProps(Properties props) throws IOException {
        Properties clusterProps = readZkClusterNodeProp(ZkUtils.getVmsClusterPath(props));
        Properties mergeProps = new Properties();
        mergeProps.putAll(clusterProps);
        mergeProps.putAll(props);
        return mergeProps;
    }

    public ChannelMetaData readChannelMetaData(String channelName, String clientId) throws IOException {
        Properties mergeProps = getMergePropsFromZkNodePath(channelName, ZkNodeType.CHANNEL);
        return new ChannelMetaData(vmsClientContext.getBelongAttr(), channelName, clientId, mergeProps);
    }

    public QueueMetaData readQueueMetaData(String queueName, String clientId) throws IOException {
        Properties mergeProps = getMergePropsFromZkNodePath(queueName, ZkNodeType.QUEUE);
        return new QueueMetaData(vmsClientContext.getBelongAttr(), queueName, clientId, mergeProps);
    }

    public DlqQueueMetaData readDLQQueueMetaData(String queueName, String clientId) throws IOException {
        Properties mergeProps = getMergePropsFromZkNodePath(queueName, ZkNodeType.QUEUE);
        String tmpClientId = (clientId == null || clientId.isEmpty()) ? "DLQ_" + Utils.randomStr() : clientId;
        return new DlqQueueMetaData(vmsClientContext.getBelongAttr(), queueName, tmpClientId, mergeProps);
    }

    private QueueMetaData readQueueMetaData(Properties props, String queueName, String clientId) {
        return new QueueMetaData(vmsClientContext.getBelongAttr(), queueName, clientId, props);
    }

    private ChannelMetaData readChannelMetaData(Properties props, String channelName, String clientId) {
        return new ChannelMetaData(vmsClientContext.getBelongAttr(), channelName, clientId, props);
    }

    public List<String> getQueueShardsInfo(String queueName) {
        try {
            Set<String> shardsInfo = Utils
                    .splitShardsTagsByString(ZkUtils.getQueueShards(readZkQueueNodeProp(queueName)));
            ArrayList<String> rtnList = new ArrayList<String>(shardsInfo.size());
            for (String iShardsInfo : shardsInfo) {
                rtnList.add(iShardsInfo);
            }
            return rtnList;
        } catch (IOException e) {
            LOGGER.error("read queues shards info error", e);
            return Collections.emptyList();
        }
    }

    @Override
    public synchronized boolean init(IContext context) throws VmsClientException {
        if (!isOpen) {
            if (vmsClientContext == null) {
                throw new VmsClientException("VmsRouterManager init error, caused by null context");
            }
            newVmsZkClientByBootstrapZk();
            isOpen = true;
        }
        return true;
    }

    @Override
    public synchronized void shutdown() {
        if (isOpen) {
            try {
                vmsZkClient.close();
            } catch (Exception e) {
            }
            vmsZkClient = null;
            isOpen = false;
        }
    }
}
